package cn.caplike.demo.rabbitmq.springboot.idempotence_reliability.port.adapter.messaging.rabbitmq;

import cn.caplike.demo.rabbitmq.springboot.idempotence_reliability.application.MailSendingService;
import cn.caplike.demo.rabbitmq.springboot.idempotence_reliability.application.MessageDeliveryLogService;
import cn.caplike.demo.rabbitmq.springboot.idempotence_reliability.common.domain.model.MailRequest;
import cn.caplike.demo.rabbitmq.springboot.idempotence_reliability.domain.model.MessageDeliveryLog;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * Description: 邮件消息消费者
 *
 * @author LiKe
 * @version 1.0.0
 * @date 2020-09-02 10:13
 */
@Slf4j
@Component
public class MailRequestMessageListener {

    /**
     * 投递日志服务接口
     */
    private MessageDeliveryLogService messageDeliveryLogService;

    /**
     * 邮件发送服务接口
     */
    private MailSendingService mailSendingService;

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "${message.rabbit.mail.queue.name}", autoDelete = "true", durable = "false"),
                    exchange = @Exchange(value = "${message.rabbit.mail.exchange.name}", type = "${message.rabbit.mail.exchange.type}", autoDelete = "true", durable = "false"),
                    key = "${message.rabbit.mail.routing-key}"
            )
    )
    public void consume(
            Channel channel,
            @Payload MailRequest mailRequest, @Headers Map<String, Object> headers
    ) throws IOException {

        final String messageId = mailRequest.getMessageId();
        final MessageDeliveryLog messageDeliveryLog = messageDeliveryLogService.get(messageId);
        if (messageDeliveryLog.hasConsumed()) {
            log.info("Consumer :: 消息 ({}) 已被消费...", messageId);
            return;
        }

        // ~ Confirm
        final Long deliveryTag = MapUtils.getLong(headers, AmqpHeaders.DELIVERY_TAG);

        try {
            // ~ Send Mail
            mailSendingService.send(
                    MailRequest.builder()
                            .clientId(mailRequest.getClientId()).clientSecret(mailRequest.getClientSecret()).receiver(mailRequest.getReceiver()).build()
            );
        } catch (Exception any) {
            // ~ Mail sending fail, NACK:
            //   1. 告诉 Broker 丢弃这条消息 (default);
            //   2. 用死信队列处理 (require: requeue = false);
            //   3. 用定时任务补偿 (本例采用)
            //   4. 重回队列 (requeue = true);
            //   Reference: https://blog.csdn.net/sun_tantan/article/details/88667102
            log.error("Consumer :: mail sending failed, cause: {}", any.getMessage());
            messageDeliveryLogService.deliveryFailed(messageId);
            channel.basicNack(deliveryTag, false, false);
            return;
        }

        // ~ Mail sending success, ACK: 告诉 Broker 这条消息已经被消费
        //   TODO 当消息状态更新失败? 发送到另外一个 Exchange 进行记录, 补偿
        messageDeliveryLogService.consumedSuccess(messageId);
        channel.basicAck(deliveryTag, Boolean.FALSE);
    }

    // ~ Autowired
    // -----------------------------------------------------------------------------------------------------------------

    @Autowired
    public void setMessageDeliveryLogService(MessageDeliveryLogService messageDeliveryLogService) {
        this.messageDeliveryLogService = messageDeliveryLogService;
    }

    @Autowired
    public void setMailSendingService(MailSendingService mailSendingService) {
        this.mailSendingService = mailSendingService;
    }
}
